
package com.scut.campus.infra.mq;

import com.scut.campus.entity.Comment;
import com.scut.campus.service.CommentService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
@Component
@RocketMQMessageListener(
        topic = "Comment",
        consumerGroup = "${spring.rocketmq.consumer.group_comment}",
        nameServer = "${spring.rocketmq.name-server}",
        consumeTimeout = 15000// 超时时间（毫秒）

)
public class TimestampBasedConsumer implements RocketMQListener<Comment>,
        RocketMQPushConsumerLifecycleListener {
    private final BlockingQueue<Comment> messageBuffer = new LinkedBlockingQueue<>();

    private static final int BATCH_THRESHOLD = 50;
    private static final long TIME_WINDOW_MS = 5000;
    private static final int MSG_BATCH_SIZE = 100;
    private static volatile Long timeFlag;

    @Autowired
    private CommentService commentService;

    public TimestampBasedConsumer() {

    }

    private synchronized void processBatch(List<Comment> commentList) {
        //List<MessageExt> batch = new ArrayList<>();
        //messageBuffer.drainTo(batch);
        if (commentList.isEmpty()) {
            return;
        }
        //数据库批量插入
        commentService.saveOrUpdateBatch(commentList);
    }

    @Override
    public void onMessage(Comment comment) {
        log.info("messageCopy");
        messageBuffer.add(comment);

        if(messageBuffer.size() >= BATCH_THRESHOLD||checkTime()) {
            List<Comment> batch = new ArrayList<>();
            int size = messageBuffer.drainTo(batch); // 取出所有元素到 batch 并清空队列
            log.info("handle msg count:{}", size);
            processBatch(batch);

        }
        else{
            log.info("not satisfied");
        }


    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        timeFlag = System.currentTimeMillis();
        consumer.setConsumeMessageBatchMaxSize(BATCH_THRESHOLD);
        //拉取消息值
        consumer.setPullBatchSize(MSG_BATCH_SIZE);
        //拉取等待时间，消息不足PullBatchSize的最大等待时间
        consumer.setPullTimeDelayMillsWhenException(TIME_WINDOW_MS);
    }

    public boolean checkTime(){
        return System.currentTimeMillis() - timeFlag > TIME_WINDOW_MS;
    }


}
